Refactor stratum server leak handling#3876
Conversation
|
Heads-up: #3889 also targets #3867 with the same structural approach ( That PR additionally adds idle timeout, max workers, and bounded write queues for the half-open / reconnect-storm failure mode. Worth coordinating so the two don’t land overlapping refactors without combining the resource bounds. |
wiesche89
left a comment
wiesche89
left a comment
There was a problem hiding this comment.
I tested this with a miner and additional connection tests. The cleanup works for normal disconnects, but it does not fully solve the reported connection leak yet. I’ve left a few comments on the remaining cases.
https://github.com/GetGrin/grim/issues/19
| wallet/db | ||
| .idea/ | ||
| .vscode/ | ||
| .venv/ |
There was a problem hiding this comment.
This looks unrelated to the Stratum change. Could we leave it out of this PR?
| outgoing = rx.next() => { | ||
| match outgoing { | ||
| Some(line) => { | ||
| if let Err(e) = framed.send(line).await { |
There was a problem hiding this comment.
While framed.send() is blocked, the socket is no longer read. In my test, this kept a half closed socket alive, while the previous split reader and writer closed it immediately. Could we keep reads and writes independent?
| } | ||
|
|
||
| async fn handle_connection(socket: TcpStream, handler: Arc<Handler>) { | ||
| let (tx, mut rx) = mpsc::unbounded(); |
There was a problem hiding this comment.
Since #3889 is closed, I think the limits should be included here. Idle connections otherwise stay open indefinitely and the unbounded queue can keep growing. Could we add an idle timeout, a connection limit, and a bounded queue?
| match listener.accept().await { | ||
| Ok((socket, _)) => { | ||
| let handler = handler.clone(); | ||
| tokio::spawn(async move { |
There was a problem hiding this comment.
Could we make the listener and connection tasks cancellable and keep their handles? That would also make clean start and stop support through the Owner API possible later.
|
|
||
| impl Drop for WorkerCleanup { | ||
| fn drop(&mut self) { | ||
| self.workers.remove_worker(self.worker_id); |
There was a problem hiding this comment.
The worker is removed from the map, but its worker_stats entry remains forever. Could we remove or reuse these entries during cleanup?
| } | ||
| } | ||
|
|
||
| async fn handle_connection(socket: TcpStream, handler: Arc<Handler>) { |
There was a problem hiding this comment.
Could we add lifecycle tests for this? The existing Stratum tests only cover JSON serialization and never reach this code.
| } | ||
| Err(e) => { | ||
| error!("accept error = {:?}", e); | ||
| continue; |
There was a problem hiding this comment.
Could we add a short backoff for accept errors? With EMFILE, this loop would otherwise spin and flood the log.
| error!("accept error = {:?}", e); | ||
| continue; | ||
| } | ||
| loop { |
There was a problem hiding this comment.
This was the last use of async stream. Could we also remove the dependency from servers/Cargo.toml?
4 participants
Addressing issue: #3867
-Reworked the stratum connection handling into a single ‘handle_connection’ loop using ‘tokio::select!’ To drive reads and writes on the same ‘Frames’ socket.